package com.junmo.www.RxJava;

import io.reactivex.*;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.observables.GroupedObservable;
import org.testng.annotations.Test;

import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

/**
 * 创建操作符
 */
public class CreateTest {
    @Test
    public  void test() {

    }
    @Test
    public  void test08() {
        Observable.range(1,8).groupBy(new Function<Integer, String>() {
            @Override
            public String apply(@NonNull Integer integer) throws Exception {
                return integer%2==0?"偶数组":"奇数组";
            }
        }).subscribe(new Consumer<GroupedObservable<String, Integer>>() {
            @Override
            public void accept(GroupedObservable<String, Integer> stringIntegerGroupedObservable) throws Exception {
                System.out.println("group name :" + stringIntegerGroupedObservable.getKey());
                if(stringIntegerGroupedObservable.getKey().equals("奇数组")){
                    stringIntegerGroupedObservable.subscribe(new Consumer<Integer>() {
                        @Override
                        public void accept(Integer integer) throws Exception {
                            System.out.println( stringIntegerGroupedObservable.getKey() + " member : " + integer) ;
                        }
                    });
                }
            }
        });
    }


    @Test
    public  void test07() throws InterruptedException {
        Observable.timer(5, TimeUnit.SECONDS).subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        System .out. println(aLong) ;
                    }
                });
        Thread.sleep(10000L);//不能让主线程停止
    }

    @Test
    public  void test06() throws InterruptedException {
        Observable.interval(1,TimeUnit.SECONDS)
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        System .out. println(aLong) ;
                    }
                });
        Thread.sleep(10000L);//不能让主线程停止
    }

    @Test
    public  void test05() {
        Observable observable = Observable.defer(new Callable<ObservableSource<?>>() {
            @Override
            public ObservableSource<?> call() throws Exception {
                return Observable . just (" hello defer " );
            }
        });
        observable.subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System.out.println (s) ;
            }
        });
        observable.subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System.out.println (s) ;
            }
        });
        observable.subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System.out.println (s) ;
            }
        });

    }

    @Test
    public void test04() {
        Observable.range(0, 9)
         .repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(@NonNull Observable<Object> objectObservable) throws Exception {
                return Observable.timer(10 , TimeUnit.SECONDS);
            }
        }).subscribe(
                onNext -> {
                    System.out.println(onNext);
                }, onError -> {
                    System.out.println(onError);
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        System.out.println("onComplete");
                    }
                });
    }

    @Test
    public  void test03() {
        Observable.just(" Hello World").repeat(10).subscribe(
                onNext -> {
                    System.out.println("onNext");
                }, onError -> {
                    System.out.println("onError");
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        System.out.println("onComplete");
                    }
                });
    }

    @Test
    public  void test02() {
        Observable.just(" Hello World").subscribe(
                onNext -> {
                    System.out.println("onNext");
                }, onError -> {
                    System.out.println("onError");
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        System.out.println("onComplete");
                    }
                });
    }

    @Test
    public  void test01() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                e.onNext (" Hello World");
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System. out . println(s) ;
            }
        });
    }
}
